package com.zhengb.flink.study.datastream.window;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Created by zhengbo on 2020/1/6.
 */
@Slf4j
public class TumblingWindowDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //每5s统计一次单词出现次数(以,分隔)
        //5s创建一个窗口 每个窗口相互独立统计 上一个5s的窗口统计不影响下一个窗口统计
        env.socketTextStream("10.0.10.36", 9999).flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {

                String[] split = value.split(",");
                for (String s : split) {
                    if (s.length() > 0) {
                        out.collect(new Tuple2<>(s, 1));
                    }
                }
            }
            //指定时间5s 在不指定时间的情况下StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic)
            //默认使用TumblingProcessingTimeWindows
        }).keyBy(0).timeWindow(Time.seconds(5)).sum(1).addSink(new SinkFunction<Tuple2<String, Integer>>() {
            @Override
            public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
                log.info("sink word count data:{}", value);
            }
        });

        env.execute("TumblingWindowDemo");

    }
}
